package mq

import (
	"encoding/json"
	"fmt"
	"github.com/streadway/amqp"
	"log"
	"logstash/config"
	"logstash/models"
	"logstash/mq/queue"
	"logstash/utile"
	"time"
)

type RmqConsumer struct {
	connection   *amqp.Connection
	channel      *amqp.Channel
	QueueName    string // 队列名称
	RoutingKey   string // key名称
	ExchangeName string // 交换机名称
	ExchangeType string // 交换机类型
	Id           string //
}

func (c *RmqConsumer) SetId() {
	c.Id = time.Now().Format("2006-01-02 15:04:05")
}

func (c *RmqConsumer) GetId() string {
	return c.Id
}

func (c *RmqConsumer) Consume() error {
	defer func() {
		if err := recover(); err != nil {
			time.Sleep(3 * time.Second)
			fmt.Printf("Consumer [%s:%s] 休息3秒,err:%s\n", c.QueueName, c.Id, err)
			c.close()
			go c.Consume()
		}
	}()
	if c.channel == nil || c.channel.Close() != nil {
		//fmt.Println("消费队列.....重连channel")
		if err := c.connect(); err != nil {
			panic(err)
		}
	}
	go c.closedConnectionListener(c.connection.NotifyClose(make(chan *amqp.Error)))

	//用于检查队列是否存在,已经存在不需要重复声明
	q, err := c.channel.QueueDeclarePassive(c.QueueName, true, false, false, true, nil)
	if err != nil {
		fmt.Printf("QueueDeclarePassive失败: [ %s ] %s \n", c.QueueName, err)
		// 队列不存在,声明队列
		// name:队列名称;durable:是否持久化,队列存盘,true服务重启后信息不会丢失,影响性能;autoDelete:是否自动删除;noWait:是否非阻塞,
		// true为是,不等待RMQ返回信息;args:参数,传nil即可;exclusive:是否设置排他
		q, err = c.channel.QueueDeclare(c.QueueName, true, false, false, true, nil)
		if err != nil {
			fmt.Printf("MQ注册队列失败: [ %s ] %s \n", c.QueueName, err)
			panic(err)
		}
	}
	utile.FailOnError(err, "Failed to declare a consumer queue")
	err = c.channel.Qos(1, 0, false)
	msgList, err := c.channel.Consume(
		q.Name, // queue
		"",     // consumer
		false,  // auto-ack
		false,  // exclusive
		false,  // no-local
		false,  // no-wait
		nil,    // args
	)
	if err != nil {
		errinfo := fmt.Sprintf("CRITICAL: Unable to start consumer %s, err : %v", c.QueueName, err)
		//log.Println(errinfo)
		fmt.Println(errinfo)
		panic(errinfo)
	}

	log.Println("[", c.QueueName, "] Running ...")
	log.Println("[", c.QueueName, "] Press CTRL+C to exit ...")
	for msg := range msgList {
		log.Println("[", c.QueueName, "] Consumed:", string(msg.Body))
		switch msg.RoutingKey {
		case queue.CONSUMER_HEARTBEAT:
			// 确认消息,必须为false
			err = msg.Ack(false)
		default:
			// 确认消息,必须为false
			err = msg.Ack(false)
			if err != nil {
				fmt.Printf("确认消息完成异常:%s \n", err)
				return err
			}
			var msgBag = &models.MessageBag{}
			err = json.Unmarshal(msg.Body, msgBag)
			if err != nil {
				fmt.Printf("json Unmarshal msgBag failed: %s \n", err)
				return err
			}
			err = utile.SendData2Es(msgBag)
			if err == nil {
				fmt.Println("发送日志到ES成功")
			} else {
				fmt.Printf("发送日志到ES失败，err：%v", err)
			}
		}
	}
	c.channel = nil
	c.connection = nil
	log.Println("[", c.QueueName, "] Exiting ...")
	return nil
}

func (c *RmqConsumer) connect() error {
	conf := config.GetConfig()
	mqconf := conf.Data.Mq
	mqUrl := fmt.Sprintf("amqp://%s:%s@%s:%s/", mqconf.Username, mqconf.Password, mqconf.Host, mqconf.Port)
	conn, err := amqp.Dial(mqUrl)
	if err != nil {
		fmt.Printf("MQ打开链接失败[%s:%s] 2:%s \n", c.QueueName, c.Id, err)
		return err
	}
	c.connection = conn
	ch, err := conn.Channel()
	if err != nil {
		fmt.Printf("MQ打开管道失败:%s \n", err)
		return err
	}
	c.channel = ch
	return nil
}

func (c *RmqConsumer) closedConnectionListener(closed <-chan *amqp.Error) {
	log.Printf("INFO: Watching closed connection for %s \n", c.QueueName)

	// If you do not want to reconnect in the case of manual disconnection
	// via RabbitMQ UI or Server restart, handle `amqp.ConnectionForced`
	// error code.
	err := <-closed
	if err != nil {
		log.Printf("INFO: Closed %s connection: %v \n", c.QueueName, err.Error())
		conf := config.GetConfig()
		var i int
		for i = 0; i < conf.Data.Mq.MaxReconnectNum; i++ {
			log.Printf("INFO: Attempting to reconnect for  %s \n", c.QueueName)

			if err := c.connect(); err == nil {
				log.Printf("INFO: Reconnected %s \n", c.QueueName)

				if err := c.Consume(); err == nil {
					break
				}
			}

			time.Sleep(time.Second * conf.Data.Mq.WaitReconnectTime)
		}

		if i == 3 {
			log.Printf("CRITICAL: Giving up reconnecting for  %s \n", c.QueueName)
			return
		}
	} else {
		log.Printf("INFO: Connection closed normally, will not reconnect for  %s \n", c.QueueName)
		//os.Exit(0)
	}
}

func (c *RmqConsumer) getName() string {
	return c.QueueName
}

func (c *RmqConsumer) close() error {
	if c.channel == nil {
		return nil
	}

	if c.connection.IsClosed() {
		return nil
	}
	// 先关闭管道,再关闭链接
	err := c.channel.Close()
	if err != nil {
		fmt.Printf("MQ管道关闭失败 2:%s \n", err)
	}
	err = c.connection.Close()
	if err != nil {
		fmt.Printf("MQ链接关闭失败 2:%s \n", err)
		return err
	}
	return nil
}
